package com.stars.easyms.mq.core;

import com.alibaba.fastjson.JSON;
import com.stars.easyms.mq.dto.MQInfoDTO;
import com.stars.easyms.mq.dto.MQSendResult;
import com.stars.easyms.redis.exception.DistributedLockTimeoutException;
import com.stars.easyms.redis.lock.DistributedLock;
import com.stars.easyms.base.batch.BatchResult;
import com.stars.easyms.mq.constant.MQConstants;
import com.stars.easyms.mq.dto.MQMessageDTO;
import com.stars.easyms.mq.enums.MQType;
import com.stars.easyms.mq.service.MQMessageService;
import com.stars.easyms.base.util.*;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.Method;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * MQ消息调度者线程(发送、接收各一个单例线程)：
 * 1.生成发送/接收处理分类者线程的线程池
 * 2.抓取发送/接收处理失败的数据
 *
 * @author guoguifang
 * @date 2018-04-23 13:54
 * @since 1.0.0
 */
@Slf4j
public final class MQScheduler implements Runnable {

    private static MQScheduler mqSendScheduler;

    private static MQScheduler mqReceiveScheduler;

    private static ThreadPoolExecutor mqSendAssorterThreadPool;

    private static ThreadPoolExecutor mqReceiveAssorterThreadPool;

    private static MQMessageService mqMessageService;

    private static MQMessageSenderFactory mqMessageSenderFactory;

    private static String applicationId;

    private static final int PER_CATCH_SIZE = 3000;

    private MQType type;

    private DistributedLock schedulerLock;

    private static final AtomicInteger LOOP_EXECUTE_EXCEPTION_COUNT = new AtomicInteger();

    public static void schedule(MQType type, List<MQMessageDTO> mqMessageDTOList) {
        try {
            if (MQType.SEND == type) {
                if (mqSendAssorterThreadPool == null) {
                    log.error("MQ服务未开启，请检查是否激活MQ服务!");
                    mqMessageService.batchUpdateMQMessageStatus(mqMessageDTOList, MQConstants.MQ_STATUS_FAIL_RETRY, null, false);
                    return;
                }
                mqSendAssorterThreadPool.execute(new MQAssorter(type, mqMessageDTOList));
            } else {
                mqReceiveAssorterThreadPool.execute(new MQAssorter(type, mqMessageDTOList));
            }
        } catch (Exception e) {
            if (log.isWarnEnabled()) {
                log.warn("线程池'mq-{}-assorter-thread-pool'队列已满!", type.getName());
            }
            // 如果分配队列已满时，把这些数据的状态改成发送失败/处理失败状态，但是失败次数不加1
            mqMessageService.batchUpdateMQMessageStatus(mqMessageDTOList, MQConstants.MQ_STATUS_FAIL_RETRY, null, true);
        }
    }

    @Override
    public void run() {
        // 设置固定参数
        Map<String, Object> paramMap = new HashMap<>(8);
        paramMap.put("failRetryStatus", MQConstants.MQ_STATUS_FAIL_RETRY);
        paramMap.put("perCatchSize", PER_CATCH_SIZE);
        paramMap.put("type", type.getCode());
        if (MQType.SEND == type) {
            paramMap.put("sourceSys", applicationId);
        } else {
            paramMap.put("destSys", applicationId);
        }

        // 循环执行mq调度任务,当失败次数过多时停止执行
        boolean isNeedPreQuery = true;
        while (LOOP_EXECUTE_EXCEPTION_COUNT.get() < MQConstants.MAX_SCHEDULE_EXCEPTION_COUNT) {
            try {
                // 1.抓取待发送或发送失败的数据
                long catchCount = catchAndScheduleData(paramMap, isNeedPreQuery);

                // 2.如果抓取数量和最大抓取数量相同则没有等待时间，否则等待30秒
                if (catchCount >= 0 && catchCount < PER_CATCH_SIZE) {
                    isNeedPreQuery = true;
                    Thread.sleep(30000 + ThreadLocalRandom.current().nextLong(5000));
                } else if (catchCount == -1) {
                    isNeedPreQuery = true;
                    Thread.sleep(1500);
                } else {
                    isNeedPreQuery = false;
                }
            } catch (Exception e) {
                log.error("MQScheduler execute failure!", e);
                LOOP_EXECUTE_EXCEPTION_COUNT.incrementAndGet();
            }
        }
    }

    private long catchAndScheduleData(Map<String, Object> paramMap, boolean isNeedPreQuery) {
        boolean flag = true;
        try {
            // 1.为了防止分布式服务器同时读取同一份数据,因此需要加锁,如果加锁失败则每隔一秒钟重新加锁
            schedulerLock.blockLock(1200000, 1000);

            // 2.判断是否需要预查询
            Long preQueryCount = null;
            if (isNeedPreQuery) {
                preQueryCount = mqMessageService.getMQMessageCount(paramMap);
            }

            // 3.根据不同的类型获取对应的消息list
            long catchCount = 0;
            if (preQueryCount == null || preQueryCount > 0) {
                List<MQMessageDTO> mqMessageDTOList = mqMessageService.getMQMessageForFailRetry(paramMap);
                if (mqMessageDTOList != null && (catchCount = mqMessageDTOList.size()) > 0) {
                    // 4.把这些数据的状态改成发送中/处理中状态
                    mqMessageService.batchUpdateMQMessageStatus(mqMessageDTOList, MQConstants.MQ_STATUS_PROCESSING, null, false);

                    // 5.分配任务给线程池，执行任务
                    schedule(type, mqMessageDTOList);
                }
            }
            return catchCount;
        } catch (DistributedLockTimeoutException e) {
            flag = false;
            log.error(e.getMessage());
        } finally {
            if (flag) {
                schedulerLock.unlock();
            }
        }
        return -1;
    }

    public static MQScheduler newMQSendSchedule(int sendAssorterCorePoolSize, int sendAssorterMaxPoolSize) {
        MQScheduler localMqSendScheduler = mqSendScheduler;
        if (localMqSendScheduler == null) {
            synchronized (MQScheduler.class) {
                localMqSendScheduler = mqSendScheduler;
                if (localMqSendScheduler == null) {
                    mqSendScheduler = localMqSendScheduler = new MQScheduler(MQType.SEND);
                    mqSendAssorterThreadPool = new ThreadPoolExecutor(sendAssorterCorePoolSize, sendAssorterMaxPoolSize, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10000),
                            new DefaultThreadFactory().setNameFormat("easyms-mq-send-assorter-%d").build(), new ThreadPoolExecutor.CallerRunsPolicy());
                    MQAssorter.mqSingleSendHandlerThreadPool = new ThreadPoolExecutor(sendAssorterCorePoolSize * 2, sendAssorterMaxPoolSize * 2, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10000),
                            new DefaultThreadFactory().setNameFormat("easyms-mq-single-send-handler-%d").build(), new ThreadPoolExecutor.CallerRunsPolicy());
                    MQAssorter.mqBatchSendHandlerThreadPool = new ThreadPoolExecutor(sendAssorterCorePoolSize * 2, sendAssorterMaxPoolSize * 2, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10000),
                            new DefaultThreadFactory().setNameFormat("easyms-mq-batch-send-handler-%d").build(), new ThreadPoolExecutor.CallerRunsPolicy());
                }
            }
        }
        return localMqSendScheduler;
    }

    public static MQScheduler newMqReceiveScheduler(int receiveAssorterCorePoolSize, int receiveAssorterMaxPoolSize) {
        MQScheduler localMqReceiveScheduler = mqReceiveScheduler;
        if (localMqReceiveScheduler == null) {
            synchronized (MQScheduler.class) {
                localMqReceiveScheduler = mqReceiveScheduler;
                if (localMqReceiveScheduler == null) {
                    mqReceiveScheduler = localMqReceiveScheduler = new MQScheduler(MQType.RECEIVE);
                    mqReceiveAssorterThreadPool = new ThreadPoolExecutor(receiveAssorterCorePoolSize, receiveAssorterMaxPoolSize, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10000),
                            new DefaultThreadFactory().setNameFormat("easyms-mq-receive-assorter-%d").build(), new ThreadPoolExecutor.CallerRunsPolicy());
                    MQAssorter.mqSingleReceiveHandlerThreadPool = new ThreadPoolExecutor(receiveAssorterCorePoolSize * 2, receiveAssorterMaxPoolSize * 2, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10000),
                            new DefaultThreadFactory().setNameFormat("easyms-mq-single-receive-handler-%d").build(), new ThreadPoolExecutor.CallerRunsPolicy());
                    MQAssorter.mqBatchReceiveHandlerThreadPool = new ThreadPoolExecutor(receiveAssorterCorePoolSize * 2, receiveAssorterMaxPoolSize * 2, 30, TimeUnit.SECONDS, new LinkedBlockingQueue<>(10000),
                            new DefaultThreadFactory().setNameFormat("easyms-mq-batch-receive-handler-%d").build(), new ThreadPoolExecutor.CallerRunsPolicy());
                }
            }
        }
        return localMqReceiveScheduler;
    }

    public static void setMQMessageService(MQMessageService mqMessageService) {
        MQScheduler.mqMessageService = mqMessageService;
    }

    public static void setApplicationId(String applicationId) {
        MQScheduler.applicationId = applicationId;
    }

    public static void setMqMessageSenderFactory(MQMessageSenderFactory mqMessageSenderFactory) {
        MQScheduler.mqMessageSenderFactory = mqMessageSenderFactory;
    }

    private MQScheduler(MQType type) {
        this.type = type;
        this.schedulerLock = new DistributedLock(MQConstants.MQ_SCHEDULER_LOCK, applicationId + ":" + type.getCode(), 60);
    }

    /**
     * MQ消息分类者线程
     */
    @Slf4j
    private static final class MQAssorter implements Runnable {

        private static ThreadPoolExecutor mqSingleSendHandlerThreadPool;

        private static ThreadPoolExecutor mqBatchSendHandlerThreadPool;

        private static ThreadPoolExecutor mqSingleReceiveHandlerThreadPool;

        private static ThreadPoolExecutor mqBatchReceiveHandlerThreadPool;

        private MQType type;

        private List<MQMessageDTO> mqMessageDTOList;

        @Override
        public void run() {
            // 根据MQ类型选择对应的线程池
            ExecutorService mqBatchHandlerThreadPool;
            if (MQType.SEND == type) {
                mqBatchHandlerThreadPool = mqBatchSendHandlerThreadPool;
            } else {
                mqBatchHandlerThreadPool = mqBatchReceiveHandlerThreadPool;
            }

            // 对list进行遍历分类
            Map<String, List<MQMessageDTO>> assortedMap = new HashMap<>(64);
            for (MQMessageDTO mqMessageDTO : mqMessageDTOList) {
                String currentKey = mqMessageDTO.getKey();
                // 判断发送是否使用批量
                MQInfoDTO mqInfoDTO;
                if (MQType.SEND == type) {
                    mqInfoDTO = MQConstants.MQ_SENDER_CHANNEL_INFO_MAP.get(currentKey);
                } else {
                    mqInfoDTO = MQConstants.MQ_RECEIVER_CHANNEL_INFO_MAP.get(currentKey);
                    if (mqInfoDTO == null) {
                        log.error("未找到有效的MQ注册信息[type:{}, key:{}]!", type, currentKey);
                        mqMessageService.updateMQMessageStatus(mqMessageDTO, MQConstants.MQ_STATUS_FAIL_RETRY, 1L, false);
                        continue;
                    }
                }
                boolean isBatch = false;
                Integer channelSize = 1;
                if (mqInfoDTO != null) {
                    isBatch = mqInfoDTO.getBatch();
                    channelSize = mqInfoDTO.getChannelSize();
                }
                if (isBatch) {
                    List<MQMessageDTO> assortedList = assortedMap.get(currentKey);
                    if (assortedList == null) {
                        assortedList = new ArrayList<>();
                        assortedMap.put(currentKey, assortedList);
                    } else if (MQType.SEND == type && assortedList.size() == channelSize) {
                        List<MQMessageDTO> copyList = new ArrayList<>(channelSize);
                        copyList.addAll(assortedList);
                        assort(mqBatchHandlerThreadPool, currentKey, null, copyList);
                        assortedList.clear();
                    }
                    assortedList.add(mqMessageDTO);
                } else {
                    if (MQType.SEND == type) {
                        assort(mqSingleSendHandlerThreadPool, currentKey, mqMessageDTO, null);
                    } else {
                        assort(mqSingleReceiveHandlerThreadPool, currentKey, mqMessageDTO, null);
                    }
                }
            }
            assortedMap.forEach((currentKey, currentValue) -> assort(mqBatchHandlerThreadPool, currentKey, null, currentValue));
        }

        private void assort(ExecutorService mqHandlerThreadPool, String currentKey, MQMessageDTO mqMessageDTO, List<MQMessageDTO> mqMessageDTOList) {
            try {
                if (mqMessageDTO != null) {
                    mqHandlerThreadPool.execute(new MQHandler(type, currentKey, mqMessageDTO));
                } else {
                    mqHandlerThreadPool.execute(new MQHandler(type, currentKey, mqMessageDTOList));
                }
            } catch (Exception e) {
                if (log.isWarnEnabled()) {
                    log.warn("线程池'mq-{}-{}-handler-thread-pool'队列已满!", mqMessageDTO != null ? "single" : "batch", type.getName());
                }
                // 如果分配队列已满时，把这些数据的状态改成发送失败/处理失败状态，但是失败次数不加1
                if (mqMessageDTO != null) {
                    mqMessageService.updateMQMessageStatus(mqMessageDTO, MQConstants.MQ_STATUS_FAIL_RETRY, null, true);
                } else {
                    mqMessageService.batchUpdateMQMessageStatus(mqMessageDTOList, MQConstants.MQ_STATUS_FAIL_RETRY, null, true);
                }
            }
        }

        private MQAssorter(MQType type, List<MQMessageDTO> mqMessageDTOList) {
            this.type = type;
            this.mqMessageDTOList = mqMessageDTOList;
        }

        /**
         * MQ发送执行者
         */
        @Slf4j
        private static final class MQHandler implements Runnable {

            private MQType type;

            private String key;

            private MQMessageDTO mqMessageDTO;

            private List<MQMessageDTO> mqMessageDTOList;

            @Override
            public void run() {
                if (MQType.SEND == type) {
                    sendHandle();
                } else {
                    receiveHandle();
                }
            }

            private void sendHandle() {
                MQInfoDTO mqInfoDTO = MQConstants.MQ_SENDER_CHANNEL_INFO_MAP.get(key);
                boolean currMessIsAllowNonPersistent = false;
                Long maxFailCount = 5L;
                if (mqInfoDTO != null) {
                    currMessIsAllowNonPersistent = mqInfoDTO.getAllowNonPersistent();
                    maxFailCount = mqInfoDTO.getMaxFailCount();
                }
                if (mqMessageDTO != null) {
                    String message = mqMessageDTO.getMessage();
                    String destinationType = mqMessageDTO.getDestinationType();
                    try {
                        MQSendResult sendResult = mqMessageSenderFactory.getMQMessageSender().convertAndSend(key, message, destinationType, currMessIsAllowNonPersistent);
                        if (sendResult.isSendSuccess()) {
                            if (log.isDebugEnabled()) {
                                log.debug("发送MQ消息到目标({})成功,消息模式:{},消息内容:{}", key, destinationType, message);
                            }
                            mqMessageService.updateMQMessageSendStatus(mqMessageDTO, sendResult, MQConstants.MQ_STATUS_SUCCESS, null, false);
                        } else {
                            log.error("发送MQ消息到目标({})失败,消息模式:{},消息内容:{}", key, destinationType, message);
                            mqMessageService.updateMQMessageSendStatus(mqMessageDTO, sendResult, MQConstants.MQ_STATUS_FAIL_RETRY, maxFailCount, false);
                        }
                    } catch (Exception e) {
                        log.error("发送MQ消息到目标({})失败,消息模式:{},消息内容:{}", key, destinationType, message, e);
                        mqMessageService.updateMQMessageStatus(mqMessageDTO, MQConstants.MQ_STATUS_FAIL_RETRY, maxFailCount, false);
                    }
                } else {
                    String message = JSON.toJSONString(mqMessageDTOList);
                    String destinationType = mqMessageDTOList.get(0).getDestinationType();
                    try {
                        MQSendResult sendResult = mqMessageSenderFactory.getMQMessageSender().convertAndSend(key, message, destinationType, currMessIsAllowNonPersistent);
                        if (sendResult.isSendSuccess()) {
                            if (log.isDebugEnabled()) {
                                log.debug("批量发送MQ消息到目标({})成功,消息模式:{},消息内容:{}", key, destinationType, message);
                            }
                            mqMessageService.batchUpdateMQMessageSendStatus(mqMessageDTOList, sendResult, MQConstants.MQ_STATUS_SUCCESS, null, false);
                        } else {
                            log.error("批量发送MQ消息到目标({})失败,消息模式:{},消息内容:{}", key, destinationType, message);
                            mqMessageService.batchUpdateMQMessageSendStatus(mqMessageDTOList, sendResult, MQConstants.MQ_STATUS_FAIL_RETRY, maxFailCount, false);
                        }
                    } catch (Exception e) {
                        log.error("批量发送MQ消息到目标({})失败,消息模式:{},消息内容:{}", key, destinationType, message, e);
                        mqMessageService.batchUpdateMQMessageStatus(mqMessageDTOList, MQConstants.MQ_STATUS_FAIL_RETRY, maxFailCount, false);
                    }
                }
            }

            @SuppressWarnings("unchecked")
            private void receiveHandle() {
                MQInfoDTO mqInfoDTO = MQConstants.MQ_RECEIVER_CHANNEL_INFO_MAP.get(key);
                Long maxFailCount = mqInfoDTO.getMaxFailCount();
                Method method = mqInfoDTO.getMethod();
                Class<?> returnType = mqInfoDTO.getReturnType();
                if (mqMessageDTO != null) {
                    String message = mqMessageDTO.getMessage();
                    Class<?> parameterClass = mqInfoDTO.getParameterClass();
                    try {
                        Object messageObj;
                        if (String.class.isAssignableFrom(parameterClass)) {
                            messageObj = message;
                        } else {
                            Type parameterType = mqInfoDTO.getParameterType();
                            if (!(parameterType instanceof ParameterizedType)) {
                                messageObj = JSON.parseObject(message, parameterClass);
                            } else {
                                messageObj = GenericTypeUtil.parseObject(message, (ParameterizedType) parameterType, parameterClass);
                            }
                        }
                        Object obj = method.invoke(mqInfoDTO.getInstance(), messageObj);
                        if (void.class.equals(returnType) || Void.class.equals(returnType)
                                || (obj instanceof Boolean && Boolean.TRUE.equals(obj))) {
                            if (log.isDebugEnabled()) {
                                log.debug("接收到目标({})MQ消息处理成功,消息内容:{}", key, message);
                            }
                            mqMessageService.updateMQMessageStatus(mqMessageDTO, MQConstants.MQ_STATUS_SUCCESS, null, false);
                            return;
                        }
                        log.error("接收到目标({})MQ消息处理失败,消息内容:{}", key, message);
                        mqMessageService.updateMQMessageStatus(mqMessageDTO, MQConstants.MQ_STATUS_FAIL_RETRY, maxFailCount, false);
                    } catch (Exception e) {
                        log.error("接收到目标({})MQ消息处理失败,消息内容:{}", key, message, e);
                        mqMessageService.updateMQMessageStatus(mqMessageDTO, MQConstants.MQ_STATUS_FAIL_RETRY, maxFailCount, false);
                    }
                } else {
                    try {
                        Class<?> parameterGenericClass = GenericTypeUtil.getGenericClass(mqInfoDTO.getParameterType(), 0);
                        ParameterizedType parameterType = GenericTypeUtil.getGenericType(mqInfoDTO.getParameterType(), 0);
                        List<Object> paramList = new ArrayList<>();
                        mqMessageDTOList.forEach(currentMqMessageDTO -> {
                            String message = currentMqMessageDTO.getMessage();
                            if (String.class.isAssignableFrom(parameterGenericClass)) {
                                paramList.add(message);
                            } else {
                                if (parameterType == null) {
                                    paramList.add(JSON.parseObject(message, parameterGenericClass));
                                } else {
                                    paramList.add(GenericTypeUtil.parseObject(message, parameterType, parameterGenericClass));
                                }
                            }
                        });
                        Object resultObj = method.invoke(mqInfoDTO.getInstance(), paramList);
                        if (resultObj instanceof BatchResult) {
                            BatchResult batchResult = (BatchResult) resultObj;
                            int successCount = batchResult.getSuccessCount();
                            if (successCount > 0) {
                                List<MQMessageDTO> successList = new ArrayList<>(successCount);
                                List<Integer> successDataIndexs = batchResult.getSuccessDataIndexs();
                                for (Integer index : successDataIndexs) {
                                    successList.add(mqMessageDTOList.get(index));
                                }
                                if (log.isDebugEnabled()) {
                                    log.debug("批量接收到目标({})MQ消息处理成功,消息内容:{}", key, batchResult.getSuccessDatas());
                                }
                                mqMessageService.batchUpdateMQMessageStatus(successList, MQConstants.MQ_STATUS_SUCCESS, null, false);
                            }
                            int failCount = batchResult.getFailCount();
                            if (failCount > 0) {
                                List<MQMessageDTO> failList = new ArrayList<>(failCount);
                                List<Integer> failDataIndexs = batchResult.getFailDataIndexs();
                                for (Integer index : failDataIndexs) {
                                    failList.add(mqMessageDTOList.get(index));
                                }
                                log.error("批量接收到目标({})MQ消息处理失败,消息内容:{}", key, batchResult.getFailDatas());
                                mqMessageService.batchUpdateMQMessageStatus(failList, MQConstants.MQ_STATUS_FAIL_RETRY, maxFailCount, false);
                            }
                            return;
                        } else if (void.class.equals(returnType) || Void.class.equals(returnType)
                                || (resultObj instanceof Boolean && Boolean.TRUE.equals(resultObj))) {
                            if (log.isDebugEnabled()) {
                                log.debug("批量接收到目标({})MQ消息处理成功,消息内容:{}", key, paramList);
                            }
                            mqMessageService.batchUpdateMQMessageStatus(mqMessageDTOList, MQConstants.MQ_STATUS_SUCCESS, null, false);
                            return;
                        }
                        log.error("批量接收到目标({})MQ消息处理失败,消息内容:{}", key, paramList);
                        mqMessageService.batchUpdateMQMessageStatus(mqMessageDTOList, MQConstants.MQ_STATUS_FAIL_RETRY, maxFailCount, false);
                    } catch (Exception e) {
                        log.error("批量接收到目标({})MQ消息处理失败,消息内容:{}", key, mqMessageDTOList, e);
                        mqMessageService.batchUpdateMQMessageStatus(mqMessageDTOList, MQConstants.MQ_STATUS_FAIL_RETRY, maxFailCount, false);
                    }
                }
            }

            private MQHandler(MQType type, String key, MQMessageDTO mqMessageDTO) {
                this.type = type;
                this.key = key;
                this.mqMessageDTO = mqMessageDTO;
            }

            private MQHandler(MQType type, String key, List<MQMessageDTO> mqMessageDTOList) {
                this.type = type;
                this.key = key;
                this.mqMessageDTOList = mqMessageDTOList;
            }
        }
    }
}